package sing.rabbit.provider.send;

import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Profile;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import static sing.rabbit.common.configurations.DirectConfiguration.DIRECT_ROUTING_KEY;

/**
 * 这里使用 @Scope 设为多例 是 为了解决报错：Only one ConfirmCallback is supported by each RabbitTemplate
 *
 * @author xing
 * @date 2022/9/9 1:14
 */
@Profile("correlated")
@Component
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public class DirectCorrelatedMessageSender {
    @Autowired
    public RabbitTemplate myRabbitTemplate;
    @Autowired
    public Queue directQueue;
    @Autowired
    private DirectExchange directExchange;

    /**
     * 注解的注释的翻译：用于依赖注入后执行初始化操作的方法，不能由任何参数，除非是拦截器
     */
    @PostConstruct
    private void setupCallbacks() {
        myRabbitTemplate.setConfirmCallback(((correlationData, ack, cause) -> {
            if(correlationData != null) {
                System.out.println("now in confirmCallBack, ack: " + ack + ", correlationData: " + correlationData);
            } else {
                System.out.println("correlated is null");
            }
        }));
        myRabbitTemplate.setReturnsCallback(returned -> {
            Message message = returned.getMessage();
            System.out.println("now in returnCallback, return message: " + message);
        });
    }

    public void sendForConfirm() throws ExecutionException, InterruptedException, TimeoutException {

        CorrelationData correlationData = new CorrelationData("this is a id");

        String message = "for confirm message";
        // 使用绑定了队列的 routingKey，为了只看 发送到交换机的 confirm，不看 returned
        myRabbitTemplate.convertAndSend(directExchange.getName(), DIRECT_ROUTING_KEY, message, correlationData);
        System.out.println("send message success: " + message);

        CorrelationData.Confirm confirm = correlationData.getFuture().get(10, TimeUnit.SECONDS);
        System.out.println("接收到是否发送到交换机确认回复, ack = " + confirm.isAck());

        // 由于能发送到至少一个队列，所以为 null
        ReturnedMessage returned = correlationData.getReturned();
        if (returned == null){
            System.out.println("Return is null");
        }
    }

    public void sendForReturned() throws ExecutionException, InterruptedException, TimeoutException {

        CorrelationData correlationData = new CorrelationData("Correlation for message 2");

        String message = "this is a  for returned message";
        // 随便取一个 routingKey, 交换机无法发送消息到至少一个队列，触发 returned
        myRabbitTemplate.convertAndSend(directExchange.getName(), "forReturned", message, correlationData);
        System.out.println("send message success: " + message);

        CorrelationData.Confirm confirm = correlationData.getFuture().get(10, TimeUnit.SECONDS);
        System.out.println("接收到是否发送到交换机确认回复, ack = " + confirm.isAck());

        // 交换机消息传递不出去，此时 returned 不为 null
        ReturnedMessage returned = correlationData.getReturned();
        if (returned != null){
            System.out.println("Return received: "  + returned.getMessage());
        }
    }
}
